package com.imooc.smartcabinet;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;

/**
 * 监控客户端状态
 * 
 * @author YT
 * 2020-05-08
 */
public class ClientStatus {
     public static void main(String[] args) throws MqttException {   
            String HOST = "tcp://127.0.0.1:11883";  
//            String TOPIC = "mqtt/test";
//            String TOPIC = "$SYS/broker/clients/connected";
            String TOPIC ="$SYS/brokers/+/clients/#";
            int qos = 1;
            String clientid = "subClient2";
            String userName = "admin";
            String passWord = "test";
            try {
                // host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
                MqttClient client = new MqttClient(HOST, clientid, new MemoryPersistence());
                // MQTT的连接设置
                MqttConnectOptions options = new MqttConnectOptions();
                // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
                options.setCleanSession(true);
                // 设置连接的用户名
                options.setUserName(userName);
                // 设置连接的密码
                options.setPassword(passWord.toCharArray());
                // 设置超时时间 单位为秒
                options.setConnectionTimeout(10);
                // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
                options.setKeepAliveInterval(20);
                // 设置回调函数
                client.setCallback(new MqttCallback() {

                    public void connectionLost(Throwable cause) {
                        System.out.println("connectionLost");
                    }

//                    public void messageArrived(String topic, MqttMessage message) throws Exception {
//                        System.out.println("topic:"+topic);
//                        System.out.println("Qos:"+message.getQos());
//                        System.out.println("message content:"+new String(message.getPayload()));
//                        
//                    }

                    public void deliveryComplete(IMqttDeliveryToken token) {
                        System.out.println("deliveryComplete---------"+ token.isComplete());
                    }
                    
                    public void messageArrived(String topic, MqttMessage message) throws Exception {
                        String msg = new String(message.getPayload());
//                        System.out.println("topic:"+topic);
//                        System.out.println("Qos:"+message.getQos());
                        System.out.println("Message content:"+msg); 
                        try {
                            JSONObject jsonObject = JSON.parseObject(msg);
                            String clientId = String.valueOf(jsonObject.get("clientid"));
                            if (topic.endsWith("disconnected")) {
                                 System.out.println("客户端已掉线：" +clientId);
                            } else {
                                  System.out.println("客户端已上线："+clientId);
                            }
                        } catch (JSONException e) {
                              System.out.println("JSON Format Parsing Exception : "+ msg);
                        }
                    }
                });
                client.connect(options);
                //订阅消息
                client.subscribe(TOPIC, qos);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
}